package dongshi.daddy.raft.leadercampaign;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;

// 模拟领导者选举过程
public class LeaderCampaignTest {
    private static CountDownLatch allNodeReady = new CountDownLatch(3);
    private static CountDownLatch mainNotExits = new CountDownLatch(5);
    private static CountDownLatch generateLeader = new CountDownLatch(1);

    private static String NODE_A_NAME = "nodeA";
    private static int NODE_A_PORT = 6668;
    private static String NODE_B_NAME = "nodeB";
    private static int NODE_B_PORT = 6669;
    private static String NODE_C_NAME = "nodeC";
    private static int NODE_C_PORT = 6670;
    private static String NODE_ROLE_FOLLOWER_NAME = "[跟随者]";
    private static String NODE_ROLE_CANDIDATE_NAME = "[候选者]";
    private static String NODE_ROLE_LEADER_NAME = "[领导者]";
    private static String VOTE_TO_BE_MSG = "我是节点A,选我当leader好吗？";
    private static String AGREE_TO_BE_LEADER = "OK";

    public static void main(String[] args) throws Exception {
        Thread nodeA = new Thread(() -> {
            try {
                ServerSocket nodeASocket = new ServerSocket(NODE_A_PORT);
                int timeout = 150;
                int term = 0;
                allNodeReady.countDown();
                allNodeReady.await();
//                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                System.out.println(Thread.currentThread().getName() + " 进入状态：" + NODE_ROLE_FOLLOWER_NAME + " 并开始倒计时" + timeout + "ms");
                // 模拟倒计时
                Thread.sleep(timeout);
                System.out.println(Thread.currentThread().getName() + " 倒计时结束，进入状态：" + NODE_ROLE_CANDIDATE_NAME + " 发起leader选举并修改任期为：" + ++term);
                sendMsg(VOTE_TO_BE_MSG, NODE_B_PORT);
                sendMsg(VOTE_TO_BE_MSG, NODE_C_PORT);
                new Thread(() -> {
                    try {
                        int counter = 0;
                        // 自己有一票，所以初始化为1
                        int agreeNum = 1;
                        while (true) {
                            counter++;
                            String msg = receiveMsg(nodeASocket);
                            System.out.println(NODE_A_NAME + " 收到投票回复：" + msg);
                            if (AGREE_TO_BE_LEADER.equalsIgnoreCase(msg)) agreeNum++;
                            if (counter >= 2) {
                                // 超过一半同意，则宣布成为leader
                                if (agreeNum > (3 / 2 + 1)) {
                                    System.out.println(NODE_A_NAME + " 获得超过半数选票，成为" + NODE_ROLE_LEADER_NAME + "，并开始发送心跳消息！");
                                    generateLeader.countDown();
                                }
                                break;
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_A_NAME);
        nodeA.start();

        Thread nodeB = new Thread(() -> {
            try {
                ServerSocket nodeBSocket = new ServerSocket(NODE_B_PORT);
                int timeout = 200;
                int term = 0;
                allNodeReady.countDown();
                allNodeReady.await();
//                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                // 因为节点A先到时，这里异步接收节点A的投票申请消息
                new Thread(() -> {
                    try {
                        String msg = receiveMsg(nodeBSocket);
                        System.out.println(NODE_B_NAME + " 收到投票消息：" + msg);
                        // 回复同意其成为leader
                        sendMsg(AGREE_TO_BE_LEADER, NODE_A_PORT);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();
                System.out.println(Thread.currentThread().getName() + " 进入状态：" + NODE_ROLE_FOLLOWER_NAME + " 并开始倒计时" + timeout + "ms");
                // 模拟倒计时
                Thread.sleep(timeout);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_B_NAME);
        nodeB.start();

        Thread nodeC = new Thread(() -> {
            try {
                ServerSocket nodeCSocket = new ServerSocket(NODE_C_PORT);
                int timeout = 300;
                int term = 0;
                allNodeReady.countDown();
                allNodeReady.await();
//                System.out.println(Thread.currentThread().getName() + " 准备完毕！");
                new Thread(() -> {
                    try {
                        String msg = receiveMsg(nodeCSocket);
                        System.out.println(NODE_C_NAME + " 收到投票消息：" + msg);
                        // 回复同意其成为leader
                        sendMsg(AGREE_TO_BE_LEADER, NODE_A_PORT);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();

                System.out.println(Thread.currentThread().getName() + " 进入状态：" + NODE_ROLE_FOLLOWER_NAME + " 并开始倒计时" + timeout + "ms");
                // 模拟倒计时
                Thread.sleep(timeout);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, NODE_C_NAME);
        nodeC.start();

//        mainNotExits.await();
        generateLeader.await();

    }

    public static void sendMsg(String sendMsg, int port) throws Exception {
        Socket socket = new Socket();
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }

    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

}
